package se.sics.kompics.tutorial.tutorial3.RIWC;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.HashSet;
import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import se.kth.ict.id2203.pp2p.PerfectPointToPointLink;
import se.kth.ict.id2203.pp2p.Pp2pSend;
import se.sics.kompics.ComponentDefinition;
import se.sics.kompics.Handler;
import se.sics.kompics.Kompics;
import se.sics.kompics.Negative;
import se.sics.kompics.Positive;
import se.sics.kompics.Start;
import se.sics.kompics.address.Address;
import se.sics.kompics.timer.ScheduleTimeout;
import se.sics.kompics.timer.Timer;
import se.sics.kompics.tutorial.tutorial1.appPFD.PerfectFailureDetector;
import se.sics.kompics.tutorial.tutorial1.appPFD.crash.CrashEvent;
import se.sics.kompics.tutorial.tutorial3.RIWC.event.ReadRequest;
import se.sics.kompics.tutorial.tutorial3.RIWC.event.ReadResponse;
import se.sics.kompics.tutorial.tutorial3.RIWC.event.WriteRequest;
import se.sics.kompics.tutorial.tutorial3.RIWC.event.WriteResponse;
import se.sics.kompics.tutorial.tutorial3.RIWC.message.AckMessage;
import se.sics.kompics.tutorial.tutorial3.RIWC.message.WriteMessage;
import se.sics.kompics.tutorial.tutorial3.RIWC.timeout.SleepTimeout;

public class RIWC_Impl extends ComponentDefinition {

	Positive<PerfectFailureDetector> pfd = requires(PerfectFailureDetector.class);
	Positive<PerfectPointToPointLink> pp2p = requires(PerfectPointToPointLink.class);
	Positive<Timer> timer = requires(Timer.class);
	Negative<AtomicRegister> atReg = provides(AtomicRegister.class);
	
	private String[] commands;
	private int lastCommand;
	private static final Logger logger = LoggerFactory.getLogger(RIWC_Impl.class);
	
	private Set<Address> neighborSet;
	private Set<Address> correct;
	private Address self;
	private int[] reqid, readval, v, ts, mrank;
	private boolean[] reading;
	private Set<Address>[] writeSet;
	
	public RIWC_Impl(){
		subscribe(handleInit, control);
		subscribe(handleStart, control);
		subscribe(handleCrash, pfd);
		subscribe(handleReadRequest, atReg);
		subscribe(handleWriteRequest, atReg);
		subscribe(handleWriteMessage, pp2p);
		subscribe(handleAck, pp2p);
		subscribe(handleTimeout, timer);
		
	}
	
	// Read request handler
	Handler<ReadRequest> handleReadRequest = new Handler<ReadRequest>() {

		@Override
		public void handle(ReadRequest request) {
			int r = request.getRegisterNumber();
			reqid[r] ++;
			reading[r] = true;
			writeSet[r].clear();
			readval[r] = v[r];
			doBroadcast(new WriteMessage(self, r, reqid[r], ts[r], mrank[r], v[r]));
		}
	};
	
	// Write request handler
	Handler<WriteRequest> handleWriteRequest = new Handler<WriteRequest>(){

		@Override
		public void handle(WriteRequest request) {
			int r = request.getRegisterNumber();
			reqid[r] ++;
			writeSet[r].clear();
			doBroadcast(new WriteMessage(self, r, reqid[r], ts[r]+1, self.getPort(), request.getValue()));
		}
		
	};
	
	// Receive broadcasted write indication
	Handler<WriteMessage> handleWriteMessage = new Handler<WriteMessage>() {

		@Override
		public void handle(WriteMessage message) {
			int r = message.getRegisterNumber();
			logger.info("Received write message");
			/*logger.info("State :");
			logger.info("ts = "+ts[r]);
			logger.info("mrank = "+mrank[r]);
			logger.info("Message :");
			logger.info("ts = "+message.getTs());
			logger.info("mrank = "+message.getMrank());*/
			if( 	(message.getTs() > ts[r]) || 
					(message.getTs() == ts[r] && message.getMrank() > mrank[r]) ){
				logger.info("Updating value of register "+r+" from "+v[r]+" to "+message.getV());
				v[r] = message.getV();
				ts[r] = message.getTs();
				mrank[r] = message.getMrank();

			}
			trigger(new Pp2pSend(message.getSource(),new AckMessage(self, r, message.getReqid())), pp2p);
		}
	};
	
	// Handle ACKs
	Handler<AckMessage> handleAck = new Handler<AckMessage>() {

		@Override
		public void handle(AckMessage message) {
			logger.info("Received ACK");
			if(message.getReqid() == reqid[message.getRegisterNumber()]){
				writeSet[message.getRegisterNumber()].add(message.getSource());
				// Have all correct node answered for this register ?
				doCheckConsensus(message.getRegisterNumber());
			}
		}
	};
	
	// Crash handler
	Handler<CrashEvent> handleCrash = new Handler<CrashEvent>() {

		@Override
		public void handle(CrashEvent event) {
			correct.remove(event.getCrashedNode());
			// Have all remaining node answered for a register ?
			for(int i = 0; i < reqid.length; i++){
				doCheckConsensus(i);
			}
		}
	};
	
	// Handle timeout for waking up
	Handler<SleepTimeout> handleTimeout = new Handler<SleepTimeout>(){

		@Override
		public void handle(SleepTimeout arg0) {
			doNextCommand();
		}
	};
	
	// Init Handler
	Handler<RIWC_Init> handleInit = new Handler<RIWC_Init>() {
		
		@SuppressWarnings("unchecked")
		@Override
		public void handle(RIWC_Init event) {
			commands = event.getCommandScript().split(":");
			lastCommand = -1;
			
			// Topology init
			neighborSet = event.getNeighborSet();
			correct = event.getNeighborSet();
			self = event.getSelf();
			
			// State init (with the given number of registers)
			int r = event.getNbRegister();
			reqid = new int[r]; 
			readval = new int[r];
			v = new int[r];
			ts = new int[r];
			mrank = new int[r];
			reading = new boolean[r];
			writeSet = new Set[r];
			for(int i = 0; i < r; i++){
				writeSet[i] = new HashSet<Address>();
			}
		}
	};
	
	// Check the consensus on a register
	private void doCheckConsensus(int r){
		if(writeSet[r].containsAll(correct) && writeSet[r].contains(self)){
			if(reading[r]){
				reading[r] = false;
				logger.info("Read value from register "+r+" : "+readval[r]);
				trigger(new ReadResponse(r, readval[r]), atReg);
			}
			else {
				logger.info("Wrote value on register "+r);
				trigger(new WriteResponse(r), atReg);
			}
			doNextCommand();
		}
	}
	
	private void doBroadcast(WriteMessage writeMessage) {
		for(Address node : neighborSet){
			trigger(new Pp2pSend(node, writeMessage), pp2p);
		}
		trigger(new Pp2pSend(self, writeMessage), pp2p);
	}

	Handler<Start> handleStart = new Handler<Start>() {
		public void handle(Start event) {
			logger.info("Indeed started");
			doNextCommand();
		}
	};
	
	private final void doNextCommand() {
		lastCommand++;

		if (lastCommand > commands.length) {
			return;
		}
		if (lastCommand == commands.length) {
			logger.info("DONE ALL OPERATIONS");
			Thread applicationThread = new Thread("ApplicationThread") {
				public void run() {
					BufferedReader in = new BufferedReader(
							new InputStreamReader(System.in));
					while (true) {
						try {
							String line = in.readLine();
							doCommand(line);
						} catch (Throwable e) {
							e.printStackTrace();
						}
					}
				}
			};
			applicationThread.start();
			return;
		}
		String op = commands[lastCommand];
		doCommand(op);
	}

	private void doCommand(String cmd) {
		if (cmd.startsWith("R")) {
			logger.info("Reading from register 0");
			handleReadRequest.handle(new ReadRequest(0));
		} else if (cmd.startsWith("W")) {
			logger.info("Writing "+cmd.substring(1)+" on register 0");
			handleWriteRequest.handle(new WriteRequest(0, Integer.parseInt(cmd.substring(1))));
		} else if (cmd.startsWith("D")) {
			logger.info("Sleeping for "+cmd.substring(1)+"ms");
			ScheduleTimeout requestCheck = new ScheduleTimeout(Integer.parseInt(cmd.substring(1)));
			requestCheck.setTimeoutEvent(new SleepTimeout(requestCheck));
			trigger(requestCheck, timer);
		} else if (cmd.equals("help")) {
			doHelp();
			doNextCommand();
		} else if (cmd.startsWith("X")) {
			doShutdown();
		} else {
			logger.info("Bad command: '{}'. Try 'help'", cmd);
			doNextCommand();
		}
	}

	private final void doHelp() {
		logger.info("Available commands: W<m>, R, help, X");
		logger.info("Wm: write the message 'm'");
		logger.info("R: read the register.");
		logger.info("help: shows this help message");
		logger.info("X: terminates this process");
	}
	
	private void doShutdown() {
		System.out.println("2DIE");
		System.out.close();
		System.err.close();
		Kompics.shutdown();
	}
}
